package com.yym.rxjavademo;

import android.os.Looper;
import android.util.Log;

import org.reactivestreams.Publisher;

import java.util.concurrent.atomic.AtomicInteger;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.FlowableSubscriber;
import io.reactivex.rxjava3.core.FlowableTransformer;
import io.reactivex.rxjava3.core.Maybe;
import io.reactivex.rxjava3.core.MaybeOnSubscribe;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.core.SingleObserver;
import io.reactivex.rxjava3.core.SingleOnSubscribe;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxTool {

    private static final String TAG = "RxTool";

    /**
     * 线程切换
     */
    public static void FlowableCreate() {
        Flowable.create((FlowableOnSubscribe<Integer>) emitter -> {
            Thread.sleep(5000);
            Log.d(TAG, "被观察者运行在主线程？ " + (Looper.getMainLooper() == Looper.myLooper()));
            for (int i = 0; i < 10; i++) {
                emitter.onNext(i);
            }
            emitter.onComplete();
        }, BackpressureStrategy.BUFFER)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(integer -> {
                    Log.d(TAG, "观察者接收到: " + integer);
                }, throwable -> {
                    Log.e(TAG, throwable.getMessage());
                }, () -> {
                    Log.d(TAG, "观察者运行在主线程？ " + (Looper.getMainLooper() == Looper.myLooper()));
                    Log.d(TAG, "执行完成");
                });
    }

    public static void FlowableJust() {
        Flowable.just(1)
                .subscribe(integer -> {
                    Log.d(TAG, "观察者接收到: " + integer);
                }, throwable -> {
                    Log.e(TAG, throwable.getMessage());
                });
    }

    public static void FlowableRange() {
        //发送一组序列数据
        Flowable.range(0, 5)
                .subscribe(integer -> {
                    Log.d(TAG, "观察者接收到: " + integer);
                }, throwable -> {
                    Log.e(TAG, throwable.getMessage());
                });
    }

    public static void FlowableSingle() {
        //Single只发射单个数据或错误事件，即使发射多个数据，后面发射的数据也不会处理。
        Single.create((SingleOnSubscribe<Integer>) emitter -> {
            emitter.onError(new Exception("抛出异常"));
            emitter.onSuccess(12);
        }).subscribe(integer -> {
            Log.d(TAG, "观察者接收到: " + integer);
        }, throwable -> {
            Log.e(TAG, throwable.getMessage());
        });
    }

    public static void FlowableMaybe() {
        //Maybe 也只能发射单个事件或错误事件，即使发射多个数据，后面发射的数据也不会处理。可以看成是Single和Completable的结合。
        Maybe.create((MaybeOnSubscribe<Integer>) emitter -> {
            emitter.onComplete();
            emitter.onSuccess(111);
        }).subscribe(integer -> {
            Log.d(TAG, "观察者接收到: " + integer);
        }, throwable -> {

        }, () -> {
            Log.d(TAG, "执行完成");
        });
    }

    /**
     * 顺序连接多个数据流，控制第一个成功不在向下执行其他数据流
     */
    public static void concat() {
        Observable<String> a1 = Observable.create((ObservableOnSubscribe<String>) emitter -> {
            emitter.onNext("a1");
            emitter.onComplete();
        });

        Observable<String> a2 = Observable.create((ObservableOnSubscribe<String>) emitter -> {
            emitter.onNext("a2");
            emitter.onComplete();
        });

        Observable<String> a3 = Observable.create((ObservableOnSubscribe<String>) emitter -> {
            emitter.onNext("a3");
            emitter.onComplete();
        });

        //Concat操作符连接多个Observable的输出，就好像它们是一个Observable，第一个Observable发射的所有数据在第二个Observable发射的任何数据前面
        Observable.concat(a1, a2, a3)
                //哪个数据源有数据，我就用哪个数据，一旦获取到想要的数据，后续数据源不再执行
                .firstElement()
                .subscribeOn(Schedulers.io())
                .subscribe(str -> {
                    Log.d(TAG, "观察者接收到: " + str);
                }, throwable -> {

                }, () -> {
                    Log.d(TAG, "执行完成");
                });
    }

    /**
     * 流并行处理
     */
    public static void map() {
        Flowable.range(1, 10)
                //flatMap运行各自独立流，合并结果；不保证任何顺序，内部流的最终结果可能最终交错；
                // concatMap 一次映射并运行一个内部流，保证顺序
                //concatMapEager 它会“同时”运行所有内部流程，但输出流程将按创建内部流程的顺序排列。
                .flatMap(v ->
                        Flowable.just(v)
                                .subscribeOn(Schedulers.io())
                                .map(w -> w * w)
                )
                .blockingSubscribe(ii -> {
                    Log.e(TAG, ii + "");
                });
        Log.e(TAG, "blockingSubscribe:阻塞执行，上面执行完才到这句");
        Log.e(TAG, "subscribe:上面没执行完就可能执行到这边");
    }

    /**
     * 依赖子流程：传递值
     */
    public static void map2() {
        Flowable.create((FlowableOnSubscribe<Integer>) emitter -> {
            Thread.sleep(5000);
            emitter.onNext(1);
            Log.e(TAG, "1. " + Thread.currentThread().getName());
        }, BackpressureStrategy.LATEST)
                .flatMap(v ->
                        Flowable.create((FlowableOnSubscribe<Integer>) e -> {
                            e.onNext(v + 1);
                            Log.e(TAG, "2. " + Thread.currentThread().getName());
                        }, BackpressureStrategy.LATEST)
                                .flatMap(vv ->
                                        Flowable.create((FlowableOnSubscribe<Integer>) e -> {
                                            e.onNext(vv + 1);
                                            Log.e(TAG, "3. " + Thread.currentThread().getName());
                                        }, BackpressureStrategy.LATEST))
                )
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(integer -> {
                    Log.e(TAG, "返回最终结果:" + integer);
                    Log.e(TAG, "4. 订阅者线程" + Thread.currentThread().getName());
                });
        Log.e(TAG, "map2()执行完成 ");
    }

    /**
     * 延续：传递值
     */
    public static void map3() {
        Flowable.just(1)
                .flatMap(integer -> {
                    Log.e(TAG, "1: " + integer);
                    return Flowable.just(integer + 1);
                })
                .flatMap(integer -> {
                    Log.e(TAG, "2: " + integer);
                    return Flowable.just(integer + 2);
                })
                .flatMap(integer -> {
                    Log.e(TAG, "3: " + integer);
                    return Flowable.just(integer + 5);
                })
                .subscribe(integer -> Log.e(TAG, "最终结果: " + integer));
    }

    /**
     * 延迟依赖
     */
    public static final void andthen() {
        AtomicInteger count = new AtomicInteger();
        Observable.range(1, 10)
                .doOnNext(ignored -> count.incrementAndGet())
                .ignoreElements()
                .andThen(Single.defer(() -> Single.just(count.get())))
//                .andThen(Single.fromCallable(() -> count.get()))
                .subscribe(integer -> Log.e(TAG, "" + integer));
    }
}
